package cn.ipanel.bigdata.job.basic

import cn.ipanel.bigdata.boot.Job
import cn.ipanel.bigdata.boot.config.{ConfigKey, Configuration}
import cn.ipanel.bigdata.dw.dim.phoenix.{T_CLASS_TAG => TG, T_DEVICE => S}
import cn.ipanel.bigdata.dw.dws.{T_DEVICE_INCR_TAG => T}
import cn.ipanel.bigdata.utils.Dictionary.{F_DATE, F_STATUS, BusState}
import org.apache.spark.sql.functions._


/**
 * 按设备类型，统计新增和删除数据
 */
class DeviceIncrByTag extends Job {

  val F_TAG = T.F_TAG

  override def onStartup(): Unit = {
    calByDay()
    calByMonth()
  }

  def calByDay(): Unit = {
    T.deleteByPeriodTime(_DAY.toSimpleDate)

    val today = _DAY.printFormat("yyyy-MM-dd")
    T.save(
      S.load
        .filter(
          substring(col(S.F_REGION), 1, 6) === lit(Configuration.getParam(ConfigKey.ROOT_AREA_CODE).substring(0, 6))
            and
          (
            // 当天新增：substring 截取，只要当天的数据，状态为新增
            (substring(col(S.F_CREATE_TIME), 1, "yyyy-MM-dd".length) === lit(today) and col(F_STATUS) === lit(BusState.ADD))
              or
              // 当天新增：substring 截取，只要当天的数据，状态为删除
              (substring(col(S.F_UPDATE_TIME), 1, "yyyy-MM-dd".length) === lit(today) and col(F_STATUS) === lit(BusState.DEL))
            )
        )
        .select(S.F_DEVICE_ID, S.F_DEVICE_CLASS, F_STATUS, S.F_REGION)
        .join(
          // 补充 TAG 字段
          TG.load
            .filter(col(F_DATE) === lit(_DAY.toDate))
            .select(TG.F_CLASS, TG.F_TAG)
            .withColumnRenamed(TG.F_CLASS, S.F_DEVICE_CLASS)
            .withColumnRenamed(TG.F_TAG, F_TAG),
          Seq(S.F_DEVICE_CLASS)
        )
        .groupBy(S.F_REGION, F_TAG)
        .agg(
          countDistinct(when(col(F_STATUS) === BusState.ADD, col(S.F_DEVICE_ID)).otherwise(null)) as T.F_ADD_NUM,
          countDistinct(when(col(F_STATUS) === BusState.DEL, col(S.F_DEVICE_ID)).otherwise(null)) as T.F_DEL_NUM
        )
        .withColumn(T.F_PERIOD_TIME, lit(_DAY.toSimpleDate))
    )
  }

  def calByMonth(): Unit = {
    val thisMonth = _DAY.asMonth
    val thisMonthDate = thisMonth.printFormat("yyyy-MM-dd")
    val nextMonthDate = thisMonth.next().printFormat("yyyy-MM-dd")
    val flag = "_FLAG"

    T.deleteByPeriodTime(_DAY.toSimpleMonth)

    T.save(
      S.load
        .filter(substring(col(S.F_REGION), 1, 6) === lit(Configuration.getParam(ConfigKey.ROOT_AREA_CODE).substring(0, 6)))
        .select(S.F_DEVICE_ID, S.F_DEVICE_CLASS, F_STATUS, S.F_REGION, S.F_CREATE_TIME, S.F_UPDATE_TIME)
        .withColumn(flag, when((substring(col(S.F_CREATE_TIME), 1, "yyyy-MM-dd".length) < lit(nextMonthDate)
          and substring(col(S.F_CREATE_TIME), 1, "yyyy-MM-dd".length) >= lit(thisMonthDate)
          and col(F_STATUS) =!= lit(BusState.DEL))
          or
          (substring(col(S.F_UPDATE_TIME), 1, "yyyy-MM-dd".length) >= lit(nextMonthDate)
            // 月增量设备，取创建时间小于下个月开始，大于等于本月开始时间（可以为编辑）；状态为销户时，销户时间要在本月之后
            and col(F_STATUS) === lit(BusState.DEL)), BusState.ADD)
          .when(substring(col(S.F_UPDATE_TIME), 1, "yyyy-MM-dd".length) < lit(nextMonthDate)
            and substring(col(S.F_UPDATE_TIME), 1, "yyyy-MM-dd".length) >= lit(thisMonthDate)
            // 月销户设备，只需要销户时间在本月即可
            and col(F_STATUS) === lit(BusState.DEL), BusState.DEL)
          .otherwise(null))
        // 过滤掉不是新增，也不是删除的设备，不纳入统计
        .filter(col(flag).isNotNull)
        .join(
          // 补充 TAG 字段
          TG.load
            .filter(col(F_DATE) === lit(_DAY.toDate))
            .select(TG.F_CLASS, TG.F_TAG)
            .withColumnRenamed(TG.F_CLASS, S.F_DEVICE_CLASS)
            .withColumnRenamed(TG.F_TAG, F_TAG),
          Seq(S.F_DEVICE_CLASS)
        )
        .groupBy(S.F_REGION, F_TAG)
        .agg(
          countDistinct(when(col(flag) === lit(BusState.ADD), col(S.F_DEVICE_ID)).otherwise(null)) as T.F_ADD_NUM,
          countDistinct(when(col(flag) === lit(BusState.DEL), col(S.F_DEVICE_ID)).otherwise(null)) as T.F_DEL_NUM
        )
        .withColumn(T.F_PERIOD_TIME, lit(thisMonth.toSimpleMonth))
    )
  }





}
